package cn.lzd.mr.case6;

import cn.lzd.mr.PathUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;

/**
 * Created by lzd on 2018/6/4.
 * 求两个用户之间共同的好友有哪些
 */
public class Case6MR {

    public static class Mp extends Mapper<LongWritable, Text, Text, Text> {

        Text k = new Text();
        Text v = new Text();

        /*
            按朋友拆，k = 朋友 ，v = 用户
            求出一个朋友对应多少用户
         */
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] userAndFriends = value.toString().split(":");

            String user = userAndFriends[0];
            String[] friends = userAndFriends[1].split(",");
            v.set(user);

            for (String f : friends) {

                k.set(f);
                context.write(k, v);
            }
        }
    }

    /*
        每个朋友下的所有用户两两组合。类似(N*(N-1))/2
        输出k = 组合用户 ，v = 朋友。
     */
    public static class Rd extends Reducer<Text, Text, Text, Text> {

        @Override
        protected void reduce(Text friend, Iterable<Text> users, Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            ArrayList<String> userList = new ArrayList<String>();

            for (Text user : users) {
                userList.add(user.toString());
            }

            Collections.sort(userList);

            System.out.println(friend + "==" + Arrays.toString(userList.toArray()));
            for (int i = 0; i < userList.size() - 1; i++) {
                for (int j = i + 1; j < userList.size(); j++) {
                    context.write(new Text(userList.get(i) + "-" + userList.get(j)), friend);
                }
            }
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        job.setJarByClass(Case6MR.class);

        job.setMapperClass(Mp.class);
        job.setReducerClass(Rd.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(PathUtil.getpath("case6") + "text"));
        FileOutputFormat.setOutputPath(job, new Path(PathUtil.getpath("case6") + "out/"));


        job.waitForCompletion(true);
    }
}
